package com.example.reactor.rsocket.controller;

import com.example.reactor.rsocket.config.entity.TimeSeriesInfo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestParam;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.util.Map;
import java.util.Optional;

/**
 * @author zpMeng
 * @since 10/11/2024
 */
@Controller
@Slf4j
public class RsocketController {

    @MessageMapping("normalRequestAndNormalResponse")
    public String normalRequestAndNormalResponse(String input) {
        return "Normal Request and Normal Response: " + input;
    }

    @MessageMapping("requestAndResponseStream")
    public Flux<String> normalRequestAndStreamResponse(@RequestParam(required = false) String user) {
        return Flux.range(1, 10)
                .map(i -> "Normal Request and Normal Response from user [%s], count: [%d]".formatted(user, i));
    }

    @MessageMapping("fireAndForget")
    public Mono<Void> fireAndForget(Mono<String> request) {
        return request
                .doOnNext(message -> System.out.println("Fire and Forget:" + message))
                .thenEmpty(Mono.empty());
    }

    @MessageMapping("monoRequestAndMonoResponse")
    public Mono<String> monoRequestAndMonoResponse(Mono<String> request) {
        return request
                .doOnNext(msg -> System.out.println("Mono Request and Mono Response: " + msg))
                .map(msg -> "Response from monoRequestAndMonoResponse: " + msg);
    }

    @MessageMapping("fluxRequestAndFluxResponse")
    public Flux<String> fluxRequestAndFluxResponse(Flux<String> request) {
        return request
                .doOnNext(msg -> System.out.println("Flux Request and Flux Response: " + msg))
                .map(msg -> "Response from fluxRequestAndFluxResponse: " + msg);
    }

    @MessageMapping("multiParam")
    public Mono<String> multiParam(String user,
                                   @Payload Mono<String> request) {

        log.info("{}", user);
        log.info("{}", request);

        return request
                .log()
                .doOnNext(msg -> log.info("MultiParam: {}", msg))
                .log()
                .map(msg -> msg + ", Response!");
    }

    @MessageMapping("multiParamFluxHeaders")
    public Flux<String> multiParamFlux(Flux<String> request,
                                       @Headers Map<String, Object> metadata) {
        log.info("multiParamFlux: {}", metadata);

        return request
                .doOnNext(item -> System.out.println("request info: " + item))
                .map(item -> "Response from multiParamFlux: " + item);
    }

    @MessageMapping("multiParamFlux")
    public Flux<String> multiParamFlux(Flux<String> request,
                                       @Header TimeSeriesInfo timeSeriesInfo) {

        Optional<Integer> reduce = timeSeriesInfo.getPayload().stream()
                .map(item -> item.getBytes().length)
                .reduce(Integer::sum);
        System.out.println(reduce.orElse(-1));

        log.info("multiParamFlux: {}", timeSeriesInfo.getId());

        return request
                .doOnNext(item -> System.out.println("request info: " + item))
                .map(item -> "Response from multiParamFlux: " + item);
    }

}